/**
 * Copyright (C) 2010 Neofonie GmbH
 *
 * This programm is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.twitterstream.sink;

import java.io.IOException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import twitter4j.Status;
import twitter4j.User;

/**
 * Class that stores tweets to HBase.
 */
public class HBaseTweetSink implements TweetSink {
  /** Logger for this class. */
  private static final Logger LOG = LoggerFactory.getLogger(HBaseTweetSink.class);

  /** Column families. */
  private static final byte[] TEXT_FAMILY = Bytes.toBytes("textFamily");
  private static final byte[] META_FAMILY = Bytes.toBytes("metaFamily");

  /** Column names. */
  private static final byte[] CREATION_DATE = Bytes.toBytes("creationDate");
  private static final byte[] FROM = Bytes.toBytes("from");
  private static final byte[] FROM_ID = Bytes.toBytes("fromId");
  private static final byte[] GEO = Bytes.toBytes("geo");
  private static final byte[] IMAGE_URL = Bytes.toBytes("imageUrl");
  private static final byte[] LANG = Bytes.toBytes("lang");
  private static final byte[] SOURCE = Bytes.toBytes("source");
  private static final byte[] TEXT = Bytes.toBytes("text");
  private static final byte[] TO = Bytes.toBytes("to");

  /** Table name. */
  private final String tableName;

  /** HBase configuration. */
  private final Configuration configuration;

  private final DateFormat dateFormat;

  /**
   * For unit tests.
   * 
   * @param tableName
   * @param cfg
   * @throws IOException
   * @throws MasterNotRunningException
   */
  HBaseTweetSink(final String tableName, Configuration cfg) throws IOException, MasterNotRunningException {
    this.tableName = tableName;
    this.configuration = cfg;
    HBaseAdmin admin = new HBaseAdmin(cfg);
    boolean found = checkTable(tableName, admin);

    if (!found) {
      initTable(tableName, admin);
    }
    this.dateFormat = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss +SSS");

  }

  public HBaseTweetSink(final String tableName) throws IOException, MasterNotRunningException {
    this.tableName = tableName;
    this.configuration = HBaseConfiguration.create();
    // Check and if needed prepare tables.
    HBaseAdmin admin = new HBaseAdmin(HBaseConfiguration.create());
    boolean found = checkTable(tableName, admin);

    if (!found) {
      initTable(tableName, admin);
    }
    this.dateFormat = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss +SSS");
  }

  private boolean checkTable(final String tablename, final HBaseAdmin admin) throws IOException {
    HTableDescriptor[] descs = admin.listTables();
    boolean found = false;
    for (HTableDescriptor desc : descs) {
      if (tablename.equals(desc.getNameAsString())) {
	found = true;
	break;
      }
    }
    return found;
  }

  private void initTable(String tablename, HBaseAdmin admin) throws IOException {
    HTableDescriptor desc = new HTableDescriptor(tablename);
    HColumnDescriptor colText = new HColumnDescriptor(TEXT_FAMILY);
    HColumnDescriptor colMeta = new HColumnDescriptor(META_FAMILY);
    desc.addFamily(colText);
    desc.addFamily(colMeta);
    if (LOG.isInfoEnabled()) {
      StringBuilder message = new StringBuilder("Creating table");
      message.append(tablename);
      message.append(" with column families " + TEXT_FAMILY + " and " + META_FAMILY + " as non were found.");
      LOG.info(message.toString());
    }
    admin.createTable(desc);
  }

  private void add(final byte[] family, final byte[] col, final String value, Put targetPut) {
    if (value != null) {
      targetPut.add(family, col, Bytes.toBytes(value));
    }
  }

  /** {@inheritDoc} */
  @Override
  public void store(final Status status) throws IOException {
    HTable table = new HTable(this.configuration, this.tableName);
    try {
      // we re-use the original tweet id from twitter as uuid
      // for more info on their generation implementation see
      // https://github.com/twitter/snowflake
      Put put = new Put(Bytes.toBytes(status.getId()));
      // format as in original tweet Fri, 17 Dec 2010 20:58:49 +0000
      add(META_FAMILY, SOURCE, status.getSource(), put);
      add(META_FAMILY, TO, "" + status.getInReplyToUserId(), put);
      add(TEXT_FAMILY, TEXT, status.getText(), put);

      if (status.getCreatedAt() != null) {
	add(META_FAMILY, CREATION_DATE, this.dateFormat.format(status.getCreatedAt()), put);
      }

      if (status.getGeoLocation() != null) {
	add(META_FAMILY, GEO, status.getGeoLocation().toString(), put);
      }

      if (status.getUser() != null) {
	User user = status.getUser();
	add(META_FAMILY, FROM, user.getName(), put);
	add(META_FAMILY, FROM_ID, "" + user.getId(), put);
	add(TEXT_FAMILY, FROM_ID, "" + user.getId(), put);
	if (user.getProfileImageURL() != null) {
	  add(META_FAMILY, IMAGE_URL, user.getProfileImageURL().toString(), put);
	}
	if (user.getLang() != null) {
	  add(META_FAMILY, LANG, user.getLang(), put);
	}
      }

      table.put(put);
      System.out.print(".");
    } catch (Throwable e) {
      // we expect there to be parsing trouble and do not want to throw away the
      // whole batch of tweets in that case.
      LOG.info(
	  "Error processing tweet: "
	      + e.getMessage()
	      + " (Ignoring and continuing processing - be alarmed if that happens often for your desired definition of often.) ",
	  e);
      LOG.info("Tweet that let to the exception: " + status);
    }
  }

}
